package com.xgq.drink.websocket;

import com.xgq.drink.config.KafkaConfig;
import com.xgq.drink.entity.ChatMessage;
import com.xgq.drink.mapper.ChatMessageMapper;
import com.xgq.drink.service.IChatMessageService;
import com.xgq.drink.utils.RedisUtil;
import com.xgq.drink.utils.TimeUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.messaging.simp.SimpMessagingTemplate;

/**
 * WebSocket聊天处理器
 */
@Slf4j
@Component
public class WebSocketChatHandler extends TextWebSocketHandler {
    @Autowired
    private RedisUtil redisUtil;
    // 存储用户ID与WebSocket会话的映射
    private static final Map<String, WebSocketSession> USER_SESSIONS = new ConcurrentHashMap<>();
    
    @Autowired
    private IChatMessageService chatMessageService;
    @Autowired
    private ChatMessageMapper chatMessageMapper;
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 建立连接后
     */
    @Override
    public void afterConnectionEstablished(@NotNull WebSocketSession session) throws Exception {
        log.info("建立连接后");
        String userId = getUserIdFromSession(session);
        if (userId != null) {
            log.info("用户[{}]建立WebSocket连接", userId);

        } else {
            log.warn("无法识别的用户建立WebSocket连接");
            session.close();
        }
    }

    /**
     * 处理消息
     */
    @Override
    protected void handleTextMessage(@NotNull WebSocketSession session, TextMessage message) {
        log.info("接收到WebSocket消息: {}", message.getPayload());
        try {
            String userId = String.valueOf(getUserIdFromSession(session));
            if (userId == null) {
                log.warn("无法识别的用户发送消息");
                return;
            }
            
            String payload = message.getPayload();
            log.info("接收到用户[{}]的消息: {}", userId, payload);
            
            // 解析消息内容
            ChatMessageRequest request = objectMapper.readValue(payload, ChatMessageRequest.class);
            
            // 根据消息类型处理
            if (request.getChatType() == 1) {
                chatMessageService.sendPrivateMessage(userId, String.valueOf(request.getReceiverId()),
                        request.getContent(), request.getMessageType());
            } else if (request.getChatType() == 2) {
                chatMessageService.sendGroupMessage(userId, String.valueOf(request.getReceiverId()),
                        request.getContent(), request.getMessageType());
            } else {
                log.warn("用户[{}]发送了无效的聊天类型: {}", userId, request.getChatType());
            }
            
        } catch (Exception e) {
            log.error("处理WebSocket消息异常", e);
        }
    }

    /**
     * 关闭连接后
     */
    @Override
    public void afterConnectionClosed(@NotNull WebSocketSession session, @NotNull CloseStatus status) {
        log.info("关闭连接后");
        String userId = getUserIdFromSession(session);
        if (userId != null) {
            log.info("用户[{}]关闭WebSocket连接", userId);
            USER_SESSIONS.remove(userId);
            // 新增离线状态更新
            new Thread(() -> {
                try {
                    Thread.sleep(5000);
                    if (!USER_SESSIONS.containsKey(userId)) {
                        redisUtil.hdel("online_users", userId);
                    }
                } catch (InterruptedException e) {
                    log.error("延迟删除线程异常", e);
                }
            }).start();
        }
    }
    
    /**
     * 从WebSocket会话中获取用户ID
     */
    private String getUserIdFromSession(WebSocketSession session) {
        Map<String, Object> attributes = session.getAttributes();
        Object userId = attributes.get("userId");
        return userId != null ? userId.toString() : null;
    }
    /**
     * 从Kafka接收私聊消息
     */
    @KafkaListener(topics = KafkaConfig.TOPIC_PRIVATE_CHAT, containerFactory = "chatMessageListenerContainerFactory")
    public void receivePrivateMessage(Map<String, Object> message) {
        System.out.println("接收到Kafka私聊消息: " + message);
        try {
            // 1. 持久化消息
            ChatMessage chatMessage = convertToEntity(message);
            chatMessageMapper.insert(chatMessage);
//            if (isUserOnline(message.get("receiverId"))) {
//                System.out.println("用户在线，发送消息: " + message);
//                messagingTemplate.convertAndSendToUser(
//                        message.get("receiverId").toString(),
//                        "/queue/private",
//                        message
//                );
//            }
            System.out.println("用户在线，发送消息: " + message);
            messagingTemplate.convertAndSendToUser(
                    message.get("receiverId").toString(),
                    "/queue/private",
                    message
            );
        } catch (Exception e) {
            log.error("处理 Kafka 私聊消息异常", e);
        }
    }
    private ChatMessage convertToEntity(Map<String, Object> message) {
        ChatMessage chatMessage = new ChatMessage();
        try {
            chatMessage.setMessageId(Long.parseLong(message.get("messageId").toString()));
            chatMessage.setSenderId(Long.parseLong(message.get("senderId").toString()));
            chatMessage.setReceiverId(Long.parseLong(message.get("receiverId").toString()));
            chatMessage.setContent(message.get("content").toString());
            chatMessage.setSendStamp(TimeUtils.getCurrentTimestamp());
            chatMessage.setMessageType((Integer) message.get("messageType"));
            chatMessage.setStatus(0); // 默认未读状态
            chatMessage.setChatType("PRIVATE_MESSAGE".equals(message.get("type")) ? 1 : 2);
            return chatMessage;
        } catch (Exception e) {
            log.error("转换消息实体失败: {}", message, e);
            return null;
        }
    }
    private boolean isUserOnline(Object receiverIdObj) {
        try {
            if (receiverIdObj == null) {
                log.warn("接收者ID为空");
                return false;
            }
            String receiverId = receiverIdObj.toString();
            System.out.println("检查用户在线状态: " + receiverId);
            System.out.println("检查用户在线状态: " + USER_SESSIONS);
            if (!USER_SESSIONS.containsKey(receiverId)) {
                System.out.println("用户不在线: " + receiverId);
            }
            return redisUtil.hasKey("ws:online:" + receiverId);
        } catch (Exception e) {
            log.error("检查用户在线状态失败", e);
            return false;
        }
    }
    /**
     * 从Kafka接收群聊消息
     */
    @KafkaListener(topics = KafkaConfig.TOPIC_GROUP_CHAT, containerFactory = "chatMessageListenerContainerFactory")
    public void receiveGroupMessage(Map<String, Object> message) {
        try {
            log.info("从Kafka接收到群聊消息: {}", message);

            String groupId = (String) message.get("groupId");
            messagingTemplate.convertAndSend("/topic/group." + groupId, message);
        } catch (Exception e) {
            log.error("处理Kafka群聊消息异常", e);
        }
    }
    
    /**
     * 聊天消息请求类
     */
    public static class ChatMessageRequest {
        private Long receiverId;
        private String content;
        private Integer messageType;
        private Integer chatType; // 1私聊, 2群聊
        
        public Long getReceiverId() {
            return receiverId;
        }
        
        public void setReceiverId(Long receiverId) {
            this.receiverId = receiverId;
        }
        
        public String getContent() {
            return content;
        }
        
        public void setContent(String content) {
            this.content = content;
        }
        
        public Integer getMessageType() {
            return messageType;
        }
        
        public void setMessageType(Integer messageType) {
            this.messageType = messageType;
        }
        
        public Integer getChatType() {
            return chatType;
        }
        
        public void setChatType(Integer chatType) {
            this.chatType = chatType;
        }
    }
} 